package com.chukun.multi.serialthreadconfinement;

import com.chukun.multi.twophase.AbstractTerminatableThread;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPClientConfig;
import org.apache.commons.net.ftp.FTPReply;

import java.io.BufferedOutputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.concurrent.*;

/**
 * @author chukun
 *  实现ftp文件的下载
 */
public class MessageFileDownloader {

    private final WorkerThread workerThread;

    public MessageFileDownloader(String outputDir,final String ftpServer,final String user,final String pwd) {
        this.workerThread = new WorkerThread(outputDir,ftpServer,user,pwd);
    }

    public void init() {
       this.workerThread.start();
    }

    public void shutdown(){
        this.workerThread.terminate();
    }

    public void downLoadFile(String file) {
        this.workerThread.download(file);
    }


    private static class WorkerThread extends AbstractTerminatableThread {

        private final BlockingQueue<String> workQueue;
        private final Future<FTPClient> ftpClientPromise;
        private final String outputDir;

        public WorkerThread(String outputDir,final String ftpServer,final String user,final String pwd) {
            this.workQueue = new ArrayBlockingQueue<>(100);
            this.outputDir = outputDir+"/";
            this.ftpClientPromise = new FutureTask<>(new Callable<FTPClient>() {
                @Override
                public FTPClient call() throws Exception {
                    FTPClient ftpClient = initFtpClient(ftpServer,user,pwd);
                    return ftpClient;
                }
            });
            // 启动线程
            new Thread((FutureTask<FTPClient>) ftpClientPromise).start();
        }

        /**
         * 下载文件
         * @param file
         */
        public void download(String file) {
            try {
                workQueue.put(file);
            }catch (InterruptedException e){
                ;
            }
        }

        /**
         * ftp客户端相关的信息初始化
         * @param ftpServer
         * @param userName
         * @param pwd
         * @throws Exception
         */
        private FTPClient initFtpClient(String ftpServer, String userName, String pwd) throws Exception {
            FTPClient ftpClient = new FTPClient();
            FTPClientConfig config = new FTPClientConfig();
            ftpClient.configure(config);
            int reply;
            ftpClient.connect(ftpServer);

            System.out.println(ftpClient.getReplyString());

            reply = ftpClient.getReply();

            if (!FTPReply.isPositiveCompletion(reply)) {
                ftpClient.disconnect();
                throw new RuntimeException("ftp server refused connection....");
            }
            boolean isOk = ftpClient.login(userName, pwd);
            if (isOk) {
                System.out.println(ftpClient.getReplyString());
            } else {
                throw new RuntimeException("fail to login.");
            }

            reply = ftpClient.cwd("~/message");
            if (!FTPReply.isPositiveCompletion(reply)) {
                ftpClient.disconnect();
                throw new RuntimeException("ftp fail to change work direction....");
            }

            ftpClient.setFileType(FTP.ASCII_FILE_TYPE);
            return ftpClient;
        }

        @Override
        protected void doRun() throws Exception {
            String file = workQueue.take();
            OutputStream os = null;
            try{
                os = new BufferedOutputStream(new FileOutputStream(outputDir+file));
                if (ftpClientPromise.isDone()) {
                    ftpClientPromise.get().retrieveFile(file,os);
                }
            }finally {
                if (null != os) {
                    os.close();
                }
                terminationToken.reservations.decrementAndGet();
            }
        }

        @Override
        protected void doCleanup(Exception ex) {
            try {
                if (ftpClientPromise.isDone()) {
                    ftpClientPromise.get().disconnect();
                }
            }catch (IOException e) {
                e.printStackTrace();
            }catch (InterruptedException e) {
                e.printStackTrace();
            }catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
}
